package cn.doitedu.rtdw.data_sync;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/3
 * @Desc: 学大数据，到多易教育
 *
 * 数据同步任务：  mysql的用户注册信息表 -->  hbase表
 **/
public class SyncJob01_MemberTable2Hbase {
    public static void main(String[] args) {

        // 创建flink编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        // 状态后端
        env.setStateBackend(new HashMapStateBackend());
        //env.setStateBackend(new EmbeddedRocksDBStateBackend());

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 用特定连接器（mysql-cdc-connector)，创建逻辑表，映射外部的真实的物理数据源 :  mysql中的 ums_member表 的binlog
        // =>逻辑表: user_mysql
        tenv.executeSql(
                "CREATE TABLE user_mysql (    " +
                        "      id BIGINT," +
                        "      username STRING,                        " +
                        "      phone STRING,                           " +
                        "      status int,                             " +
                        "      create_time timestamp(3),               " +
                        "      gender int,                             " +
                        "      birthday date,                          " +
                        "      province STRING,                        " +
                        "      city STRING,                            " +
                        "      job STRING ,                            " +
                        "      source_type INT ,                       " +
                        "     PRIMARY KEY (id) NOT ENFORCED            " +
                        "     ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw',          " +
                        "     'table-name' = 'ums_member'              " +
                        ")"
        );

        // 用特定连接器(hbase-connector)，创建逻辑表，映射外部的真实的物理的存储目标： hbase的一个表 dim_user_info
        // =>逻辑表: user_hbase
        tenv.executeSql("CREATE TABLE user_hbase( " +
                " username STRING, " +
                " f ROW<id BIGINT,phone STRING, status INT, create_time TIMESTAMP(3), gender INT, birthday DATE, province STRING, city STRING, job STRING, source_type INT>, " +
                " PRIMARY KEY (username) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_user_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");



        // 写一个sql，操作逻辑表： insert into user_hbase  select  from user_mysql
        tenv.executeSql("insert into user_hbase " +
                "select " +
                "username," +
                "row(id,phone,status,create_time,gender,birthday,province,city,job,source_type) as f " +
                "from user_mysql");

    }
}
